package hadoop.mapReduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @describe: 分区   支持对key进行分区，从而让map出来的数据均匀的分布在reducer上，
 * 框架自带的默认分区为Partitioner
 * <p>
 * 日志文件 按照级别进行分割
 */
public class Partition {

    //2020-04-14 16:34:03.800 [main] INFO
    //2020-04-14 16:34:03.800 [main] INFO
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        System.setProperty("HADOOP_USER_NAME", "root");
        String HDFS_URI = "hdfs://hdphost:9000/";
        String USER = "root";
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.set("dfs.replication", "1");
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, USER);
        Job job = Job.getInstance(conf);

        //job所在的jar包
        job.setJarByClass(Partition.class);

        job.setMapperClass(Partition1Mapper.class);
        job.setReducerClass(Partition1Reducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Bean.class);

        //设置自定义的Partitioner类，替换框架默认的hashpartitioner
        job.setPartitionerClass(LogLevelProvotioner.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Bean.class);

        //分区数需要和任务数匹配
        job.setNumReduceTasks(5);
        //hdfs的文件路径
        String inputPath = "/input";
        String outputPath = "/output";

        Path path = new Path(outputPath);
        FileSystem fileSystem = path.getFileSystem(conf);
        if (fileSystem.exists(path)) {
            fileSystem.delete(path, true);
        }
        //设置本次job的输入 输出
        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        //提交job
        job.waitForCompletion(true);
    }

    /**
     * 日志级别作为map条件
     */
    static class Partition1Mapper extends Mapper<LongWritable, Text, Text, Bean> {
        Bean tx = new Bean();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split(" ");
            Text keyout = new Text(fields[3]);
            tx.set(fields[0], fields[1], fields[4]);
            //key value
            context.write(keyout, tx);
        }
    }

    //key: level  values list<bean>
    static class Partition1Reducer extends Reducer<Text, Bean, Text, Bean> {
        @Override
        protected void reduce(Text key, Iterable<Bean> values, Context context) throws IOException, InterruptedException {
            for (Bean bean : values) {
                context.write(key, bean);
            }
        }
    }

    static class LogLevelProvotioner extends Partitioner<Text, Bean> {

        @Override
        public int getPartition(Text key, Bean value, int partitions) {
            int level = 4;

            switch (key.toString()) {
                case "INFO":
                    level = 0;
                    break;
                case "WARN":
                    level = 1;
                    break;
                case "ERROR":
                    level = 2;
                    break;
                case "DEBUG":
                    level = 3;
                    break;
                default:
                    level = 4;
            }
            return level;
        }
    }

    static class Bean implements Writable {

        //2020-04-14 16:34:03.800 [main] INFO
        private String date;
        private String time;
        private String log;

        //因为反序列化时需要反射，必须有一个构造函数，显式定义一个
        public Bean() {

        }

        public Bean(String date, String time, String log) {
            set(date, time, log);
        }

        public void set(String date, String time, String log) {
            this.date = date;
            this.time = time;
            this.log = log;
        }


        @Override
        public void write(DataOutput out) throws IOException {
            out.writeUTF(date);
            out.writeUTF(time);
            out.writeUTF(log);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            date = in.readUTF();
            time = in.readUTF();
            log =  in.readUTF();
        }

        @Override
        public String toString() {
            return date + "\t" + time + "\t" + log;
        }
    }

}
